package com.swak.reactivex.transport.codec;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.Constants;
import com.swak.exception.SerializeException;
import com.swak.reactivex.stream.Stream;
import com.swak.reactivex.transport.Codec.DecodeResult;
import com.swak.reactivex.transport.Message;
import com.swak.reactivex.transport.serialize.Serialization;
import com.swak.registry.URL;
import com.swak.spi.ExtensionLoader;
import com.swak.utils.Bytes;
import com.swak.utils.buffer.ByteBuffers;

import io.netty.buffer.ByteBuf;
import lombok.Getter;

/**
 *  通用的编解码处理器：
 *  
 *   0 1  2         3 4 5 6    7            89101112131415
 *  ------------------------------------------------------
 *   babe 00000000  00000000   00000 000    0000000000000000
 *  ------------------------------------------------------
 *   魔数  选项       长度        状态   序列化   ID（可以没有，置位 Event位（选项第3位））   
 *   
 * @author 618lf
 */
@Getter
public class MessageCodec {

	private static Logger logger = LoggerFactory.getLogger(MessageCodec.class);
	private static ExtensionLoader<Serialization> SERIALIZATION = ExtensionLoader
			.getExtensionLoader(Serialization.class);
	private static Map<Byte, Serialization> SERIALIZATION_MAP = new HashMap<Byte, Serialization>();
	private static final int SERI_MASK = 0x07;
	private static final int STATUS_MASK = 0x1f;

	static {
		List<Serialization> serializations = SERIALIZATION.loadAll(Serialization.class.getClassLoader());
		for (Serialization ser : serializations) {
			SERIALIZATION_MAP.put(ser.getType(), ser);
		}
	}

	// header length.
	protected static final int MIN_HEADER_LENGTH = 8;
	protected static final int HEADER_LENGTH = 16;
	protected static final int OPTION_POSITION = 2;
	protected static final int LEN_POSITION = 3;
	protected static final int STATUS_SERI_POSITION = 7;
	protected static final int SERI_SHIFT = 3;
	protected static final int ID_POSITION = 8;

	private final URL url;
	private final short MAGIC;
	private final byte MAGIC_HIGH;
	private final byte MAGIC_LOW;

	public MessageCodec(short MAGIC, URL url) {
		this.MAGIC = MAGIC;
		this.MAGIC_HIGH = Bytes.short2bytes(MAGIC)[0];
		this.MAGIC_LOW = Bytes.short2bytes(MAGIC)[1];
		this.url = url;
	}

	/**
	 * 编码；
	 * 
	 * @param buffer
	 * @param message
	 * @throws IOException
	 */
	public void encode(ByteBuf buffer, Message message) throws IOException {
		if (message.isRequest()) {
			this.encodeRequest(buffer, message);
		} else if (message.isResponse()) {
			this.encodeResponse(buffer, message);
		} else if (message.isEvent()) {
			this.encodeEvent(buffer, message);
		} else if (message.isStream()) {
			this.encodeStream(buffer, message);
		} else {
			this.encodeResponse(buffer, message);
		}
	}

	protected void encodeStream(ByteBuf buffer, Message message) throws IOException {

		// 数据长度
		int len = 0;

		// header.
		byte[] header = new byte[MIN_HEADER_LENGTH];

		// set magic number.
		Bytes.short2bytes(MAGIC, header);

		// set option
		header[OPTION_POSITION] = message.getOption();

		// set status（没有用）
		header[STATUS_SERI_POSITION] = message.getStatus();

		// 如果使用压缩
		if (message.isUseGzip()) {
			byte[] bytes = null;
			if (message.getData() instanceof byte[]) {
				bytes = (byte[]) message.getData();
			} else if (message.getData() instanceof ByteBuffer) {
				bytes = ByteBuffers.getBytes((ByteBuffer) message.getData());
			} else if (message.getData() instanceof ByteBuf) {
				ByteBuf data = (ByteBuf) message.getData();
				bytes = new byte[data.readableBytes()];
				data.readBytes(bytes);
				data.release();
			} else {
				throw new SerializeException("不支持此类型的流数据.");
			}
			bytes = Bytes.gzip(bytes);
			len = bytes.length;

			Bytes.int2bytes(len, header, LEN_POSITION);
			buffer.writeBytes(header);
			buffer.writeBytes(bytes);
		}

		// 不使用压缩
		else if (message.getData() instanceof byte[]) {
			byte[] bytes = (byte[]) message.getData();
			len = bytes.length;

			Bytes.int2bytes(len, header, LEN_POSITION);
			buffer.writeBytes(header);
			buffer.writeBytes(bytes);
		} else if (message.getData() instanceof ByteBuffer) {
			ByteBuffer data = (ByteBuffer) message.getData();
			len = data.limit() - data.position();

			Bytes.int2bytes(len, header, LEN_POSITION);
			buffer.writeBytes(header);
			buffer.writeBytes(data);
		} else if (message.getData() instanceof ByteBuf) {
			ByteBuf data = (ByteBuf) message.getData();
			len = data.readableBytes();

			Bytes.int2bytes(len, header, LEN_POSITION);
			buffer.writeBytes(header);
			buffer.writeBytes(data);
			data.release();
		} else if (message.getData() instanceof Stream) {
			Stream stream = (Stream) message.getData();
			len = stream.getTotalLength();

			Bytes.int2bytes(len, header, LEN_POSITION);
			buffer.writeBytes(header);
			stream.encode(buffer);
		} else {
			throw new SerializeException("不支持此类型的流数据.");
		}

		// 记录长度
		message.setHeaderLen(header.length).setBodyLen(len);
	}

	protected void encodeEvent(ByteBuf buffer, Message message) throws IOException {

		// serialization
		Serialization serialization = SERIALIZATION.load(this.getUrl().getParameter(Constants.Serialization_Key));

		// 数据长度
		int len = 0;

		// header.
		byte[] header = new byte[MIN_HEADER_LENGTH];

		// set magic number.
		Bytes.short2bytes(MAGIC, header);

		// set option
		header[OPTION_POSITION] = message.getOption();

		// set request flag and serialization type
		header[STATUS_SERI_POSITION] = (byte) (message.getStatus() << SERI_SHIFT | serialization.getType());

		// write data
		byte[] bytes = serialization.serialize(this.getUrl(), message.getData());
		if (message.isUseGzip()) {
			bytes = Bytes.gzip(bytes);
		}
		len = bytes.length;

		Bytes.int2bytes(len, header, LEN_POSITION);
		buffer.writeBytes(header);
		buffer.writeBytes(bytes);

		// 记录长度
		message.setHeaderLen(header.length).setBodyLen(len);

		/*
		 * 打印编码
		 */
		if (logger.isDebugEnabled()) {
			logger.debug("##Encode:" + Bytes.bytes2hex(header) + Bytes.bytes2hex(bytes) + "##");
		}
	}

	protected void encodeRequest(ByteBuf buffer, Message request) throws IOException {

		// serialization
		Serialization serialization = SERIALIZATION.load(this.getUrl().getParameter(Constants.Serialization_Key));

		// 数据长度
		int len = 0;

		// header.
		byte[] header = new byte[HEADER_LENGTH];

		// set magic number.
		Bytes.short2bytes(MAGIC, header);

		// set option
		header[OPTION_POSITION] = request.getOption();

		// set request flag and serialization type
		header[STATUS_SERI_POSITION] = (byte) (request.getStatus() << SERI_SHIFT | serialization.getType());

		// set request id.
		Bytes.long2bytes(request.getId(), header, ID_POSITION);

		// write data
		byte[] bytes = serialization.serialize(this.getUrl(), request.getData());
		if (request.isUseGzip()) {
			bytes = Bytes.gzip(bytes);
		}
		len = bytes.length;

		Bytes.int2bytes(len, header, LEN_POSITION);
		buffer.writeBytes(header);
		buffer.writeBytes(bytes);

		// 记录长度
		request.setHeaderLen(header.length).setBodyLen(len);

		/*
		 * 打印编码
		 */
		if (logger.isDebugEnabled()) {
			logger.debug("##Encode:" + Bytes.bytes2hex(header) + Bytes.bytes2hex(bytes) + "##");
		}
	}

	protected void encodeResponse(ByteBuf buffer, Message response) throws IOException {
		// serialization
		Serialization serialization = SERIALIZATION.load(this.getUrl().getParameter(Constants.Serialization_Key));

		// 数据长度
		int len = 0;

		// header.
		byte[] header = new byte[HEADER_LENGTH];

		// set magic number.
		Bytes.short2bytes(MAGIC, header);

		// set option
		header[OPTION_POSITION] = response.getOption();

		// set request and serialization flag.
		header[STATUS_SERI_POSITION] = (byte) (response.getStatus() << SERI_SHIFT | serialization.getType());

		// set response status.
		byte status = response.getStatus();

		// set request id.
		Bytes.long2bytes(response.getId(), header, ID_POSITION);

		// encode response data or error message.
		byte[] bytes = status == Message.OK ? serialization.serialize(this.getUrl(), response.getData())
				: serialization.serialize(this.getUrl(), response.getError());
		if (response.isUseGzip()) {
			bytes = Bytes.gzip(bytes);
		}
		len = bytes.length;

		Bytes.int2bytes(len, header, LEN_POSITION);
		buffer.writeBytes(header);
		buffer.writeBytes(bytes);

		// 记录长度
		response.setHeaderLen(header.length).setBodyLen(len);

		/*
		 * 打印编码
		 */
		if (logger.isDebugEnabled()) {
			logger.debug("##Encode:" + Bytes.bytes2hex(header) + Bytes.bytes2hex(bytes) + "##");
		}
	}

	/**
	 * 解码；
	 * 
	 * @param buffer
	 * @return
	 * @throws IOException
	 */
	public Object decode(ByteBuf buffer) {

		// check min length.
		int readable = buffer.readableBytes();
		if (readable < MIN_HEADER_LENGTH) {
			return DecodeResult.NEED_MORE_INPUT;
		}

		// read header
		byte[] header = new byte[MIN_HEADER_LENGTH];
		buffer.readBytes(header);

		// check magic number and skip noise bytes.
		if (readable > 0 && header[0] != MAGIC_HIGH || readable > 1 && header[1] != MAGIC_LOW) {
			boolean hasNoise = false;
			int length = header.length;
			if (header.length < readable) {
				header = Bytes.copyOf(header, readable);
				buffer.readBytes(header, length, readable - length);
			}
			for (int i = 1; i < header.length - 1; i++) {
				if (header[i] == MAGIC_HIGH && header[i + 1] == MAGIC_LOW) {
					buffer.readerIndex(buffer.readerIndex() - header.length + i);
					hasNoise = true;
					break;
				}
			}
			return hasNoise ? DecodeResult.SKIP_SOME_INPUT : DecodeResult.CHECK_MAGIC_ERROR;
		}

		// get option
		byte option = header[OPTION_POSITION];

		// get data length.
		int len = Bytes.bytes2int(header, LEN_POSITION);
		int tt = len + (option >> 5 > 0 ? MIN_HEADER_LENGTH : HEADER_LENGTH);
		if (readable < tt) {
			return DecodeResult.NEED_MORE_INPUT;
		}

		// decode body
		return this.decodeBody(buffer, header);
	}

	protected Message decodeBody(ByteBuf buffer, byte[] header) {

		// get flag
		byte option = header[OPTION_POSITION];
		// get status.
		byte status = (byte) ((header[STATUS_SERI_POSITION] >> SERI_SHIFT) & STATUS_MASK),
				serial = (byte) (header[STATUS_SERI_POSITION] & SERI_MASK);
		// get request id(Event 和 Stream 没有ID).
		long id = (option >> 5 > 0) ? 0L : buffer.readLong();

		// get body len
		int len = Bytes.bytes2int(header, LEN_POSITION);

		// build message
		Message message = new Message(id).setOption(option).setStatus(status)// 基本的结构
				.setHeaderLen((option >> 5 > 0) ? MIN_HEADER_LENGTH : HEADER_LENGTH) // 临时的头部长度
				.setBodyLen(len); // 数据长度

		// has no data
		if (message.isHeartbeat()) {
			if (logger.isDebugEnabled()) {
				logger.debug("##Decode:" + Bytes.bytes2hex(header) + "##");
			}
			return message;
		}

		// 流数据
		if (!message.isUseGzip() && message.isStream()) {
			message.setData(buffer.readRetainedSlice(len));
			return message;
		}

		// 非流数据 或者需要解压缩
		byte[] body = new byte[len];
		buffer.readBytes(body);

		// 解压缩
		if (message.isUseGzip()) {
			try {
				body = Bytes.unGzip(body);
			} catch (Throwable t) {
				if (logger.isWarnEnabled()) {
					logger.warn("Unzip body failed: " + t.getMessage(), t);
				}
				message.setStatus(message.isRequest() ? Message.BAD_REQUEST : Message.CLIENT_ERROR);
				message.setError((Throwable) t);
				return message;
			}
		}

		// 流数据不需要序列化
		if (message.isStream()) {
			message.setData(body);
		}
		// 使用序列化
		else {
			try {
				Serialization serialization = SERIALIZATION_MAP.get(serial);
				Object data = serialization.deserialize(this.getUrl(), body);
				if (status == Message.OK || message.isRequest()) {
					message.setData(data);
				} else {
					message.setError((Throwable) data);
				}
			} catch (Throwable t) {
				if (logger.isWarnEnabled()) {
					logger.warn("Decode message failed: " + t.getMessage(), t);
				}
				message.setStatus(message.isRequest() ? Message.BAD_REQUEST : Message.CLIENT_ERROR);
				message.setError(t);
			}
		}

		/*
		 * 打印编码
		 */
		if (logger.isDebugEnabled()) {
			logger.debug("##Decode:" + Bytes.bytes2hex(header) + Long.toHexString(id) + Bytes.bytes2hex(body) + "##");
		}
		return message;
	}
}
